package com.shujia.core

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD

object Demo18Partition {
  def main(args: Array[String]): Unit = {


    /**
     * SparkRDD的分区数
     * 1、spark读取hdfs中的数据，默认一个block对应一个分区
     * 2、读取hdfs中的数据时吗，可以改变分区数
     * 3、spark中默认后面RDD的分区数等于前面RDD的分区数
     * 4、spark中会产生shuffle类的算子可以手动指定分区数
     * 5、repartition： 重分区，底层会产生shuffle
     *
     * 分区数优化规则： 保证每一个分区的数据量在128M左右
     */

    //1、创建spark环境
    //创建配置文件对象
    val conf = new SparkConf()
    //指定spark执行默认，local：本地执行
    conf.setMaster("local")
    //spark 任务名
    conf.setAppName("wc")
    //创建spark上下文对象
    val sc = new SparkContext(conf)

    //在读取hdfs数据时可以指定一个分区的最小值，具体分区数，是通过计算得到的，spark会保证每个分区中的数据量差不多
    val lines: RDD[String] = sc.textFile("spark/data/words", 7)
    println(s"linesRDD分区数：${lines.getNumPartitions}")

    /**
     * repartition: 会产生shuffle， 可以用于增加分区，也可以用于减少分区
     * repartition： 一般用于增加分区，
     */
    // val partRDD: RDD[String] = lines.repartition(2)
    //println(s"partRDDRDD分区数：${partRDD.getNumPartitions}")

    /**
     * coalesce(2, shuffle = false): 不产生shuffle, 只能用于减少分区
     * coalesce: 一般用于合并小文件
     */
    val cosRDD: RDD[String] = lines.coalesce(2, shuffle = false)
    println(s"cosRDD分区数：${cosRDD.getNumPartitions}")

    //3、将一行转换成多行
    val words: RDD[String] = cosRDD.flatMap(line => line.split(","))
    println(s"wordsRDD分区数：${words.getNumPartitions}")


    //4、转换成kv格式
    val kvs: RDD[(String, Int)] = words.map(word => (word, 1))
    println(s"kvsRDD分区数：${kvs.getNumPartitions}")

    //5、统计单词的数量
    //reduceByKey通过key对value进行聚合计算,底层会产生shuffle
    val counts: RDD[(String, Int)] = kvs.reduceByKey((x, y) => x + y, 2)
    println(s"countsRDD分区数：${counts.getNumPartitions}")

    //6、将计算结果保存到文件中
    counts.saveAsTextFile("spark/data/word_count")


  }

}
